import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class DataSetWordCount {
    public static void main(String[] args) throws Exception {
        // 创建Flink运行的上下文环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 创建DataSet，这里我们的输入是一行一行的文本
        DataSet<String> text = env.fromElements(
                "flink spark hadoop",
                "flink flink flink",
                "spark spark hadoop",
                "flink hadoop hadoop"
        );
        // 通过Flink内置的转换函数进行计算
        DataSet<Tuple2<String, Integer>> counts = text.flatMap(new LineSplitter())
                .groupBy(0)  //0指定是按照元组中的第一个聚合
                .sum(1); //1表示按照元组中的第二个元素求和
        //打印结果
        counts.printToErr();
//        counts.print();
    }

    /**
     *  分割单词
     */
    public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            //将文本分割
            String[] words = value.toLowerCase().split("\\W+");
            for (String word : words) {
                if (word.length() > 0) {
                    out.collect(new Tuple2<String, Integer>(word, 1));
                }
            }
        }
    }
}
